package com.hu.flink12.sensor

import com.hu.entity.SensorReading
import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @Author: hujianjun
 * @Create Date: 2021/1/26 16:01
 * @Describe: 从集合读取数据
 */

object RichFlatMapSensor {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    env.setParallelism(1)
    val inputStream = env.fromCollection(List(
      SensorReading("sensor_1", 1547718199, 35.8),
      SensorReading("sensor_6", 1547718201, 15.4),
      SensorReading("sensor_7", 1547718202, 6.7),
      SensorReading("sensor_10", 1547718205, 38.1)
    ))

    inputStream.flatMap(new DefineFlatMapFunction).print()

    env.setRuntimeMode(RuntimeExecutionMode.BATCH)
    env.execute("读取传感器集合数据")
  }
}

class DefineFlatMapFunction extends RichFlatMapFunction[SensorReading, (String, String)] {

  override def open(parameters: Configuration): Unit = {
    //TODO 做一些初始化工作，比如建立数据库连接，hdfs连接
    val subtask = getRuntimeContext.getIndexOfThisSubtask
    println(s"current task id:${subtask}")
  }

  override def close(): Unit = {
    //TODO 做一些清理关闭工作，比如关闭数据库连接，hdfs连接
  }

  override def flatMap(in: SensorReading, collector: Collector[(String, String)]): Unit = {
    //返回任务id和传感器id的二元组信息
    collector.collect((getRuntimeContext.getIndexOfThisSubtask.toString, in.id))
  }
}
